package com.lu.integrate.message.center.service;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.lu.integrate.message.center.model.enums.MessageQueueLogsStatusEnum;
import com.lu.stack.common.model.response.R;
import com.lu.integrate.message.center.model.domain.MessageCenterQueueConsumeDo;
import com.lu.integrate.message.center.model.entity.SysMessageQueueConfig;
import com.lu.integrate.message.center.model.request.MessageCenterQueueConsumerRequest;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * @author zhanglu
 * @project lu-cloud-stack
 * @description 消息队列
 * @date 2023/2/13 17:47
 */
@Slf4j
@Component
public abstract class MessageQueueAbstract {

    @Resource
    private Environment env;
    @Resource
    private ThreadPoolTaskExecutor messageQueueConsumerExecutor;
    @Resource
    private SysMessageQueueConfigService messageQueueConfigService;
    @Resource
    private DbMessageQueueService dbMessageQueueService;

    @PostConstruct
    public void init(){
        dbExecutor();
    }

    /**
     * 消息消费
     * @param request
     * @return
     */
    public R consume(MessageCenterQueueConsumerRequest request){
        List<R> rs = new ArrayList<>();
        try {
            MessageQueueService messageQueueService = (MessageQueueService) SpringUtil.getBean(Class.forName(request.getType().getBeanClass()));
            QueryWrapper<SysMessageQueueConfig> configQueryWrapper = new QueryWrapper<>();
            if(ObjectUtil.isNotEmpty(request.getConfigIds())){
                configQueryWrapper.in("id", request.getConfigIds());
            }
            if(ObjectUtil.isNotEmpty(request.getKeyword())){
                configQueryWrapper.like("topic", request.getKeyword());
            }
            List<SysMessageQueueConfig> configs = messageQueueConfigService.list(configQueryWrapper);
            log.info("消息队列执行器:{}---待执行的主题数量:{}", request.getType().name(), configs.size());
            for (SysMessageQueueConfig config : configs) {
                MessageCenterQueueConsumeDo consumeDo = new MessageCenterQueueConsumeDo();
                consumeDo.setConfigId(config.getId());
                consumeDo.setClassName(config.getClassName());
                consumeDo.setMethodName(config.getMethodName());
                consumeDo.setStatus(request.getStatus());
                if(ObjectUtil.isNotEmpty(request.getBatchSize())){
                    consumeDo.setBatchSize(request.getBatchSize());
                }
                R r = messageQueueService.consume(consumeDo);
                rs.add(r);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return R.success(rs);
    }

    /**
     * db消息队列执行器
     */
    private void dbExecutor(){
        //是否启用，开启周期执行任务
        String enable = env.getProperty("message.queue.db.enable");
        if(ObjectUtil.isEmpty(enable) || !enable.equalsIgnoreCase("true")){
            log.info("db消息队列执行器-无需启动:{}", enable);
            return;
        }
        //周期执行
        new Thread(() -> {
            while (true){
                try {
                    //是否暂停
                    String stop = env.getProperty("message.queue.db.stop");
                    //处理批次大小
                    String batchSize = env.getProperty("message.queue.db.batch-size");
                    //执行等待时间s
                    String executeWaitTime = env.getProperty("message.queue.db.execute-wait-time");
                    if(ObjectUtil.isEmpty(executeWaitTime)){
                        executeWaitTime = "5";
                    }
                    //暂停等待时间s
                    String stopWaitTime = env.getProperty("message.queue.db.stop-wait-time");
                    if(ObjectUtil.isEmpty(stopWaitTime)){
                        stopWaitTime = "30";
                    }
                    if(ObjectUtil.isNotEmpty(stop) && stop.equalsIgnoreCase("true")){
                        if(log.isDebugEnabled()){
                            log.debug("db消息队列执行器-已暂停:{}---暂停等待时间:{}", stop, stopWaitTime);
                        }
                        TimeUnit.SECONDS.sleep(Integer.valueOf(stopWaitTime));
                        continue;
                    }
                    if(log.isDebugEnabled()){
                        log.debug("db消息队列执行器-开始执行---执行等待时间:{}", executeWaitTime);
                    }
                    //根据执行的方法调用分组
                    Map<String, List<SysMessageQueueConfig>> configs = messageQueueConfigService.list().stream()
                            .collect(Collectors.groupingBy(o -> o.getClassName() + o.getMethodName()));
                    if(log.isDebugEnabled()){
                        log.debug("db消息队列执行器---待执行的主题数量:{}", configs.size());
                    }
                    configs.forEach((k, v) -> {
                        messageQueueConsumerExecutor.execute(() -> {
                           v.forEach(o -> {
                               MessageCenterQueueConsumeDo consumeDo = new MessageCenterQueueConsumeDo();
                               consumeDo.setConfigId(o.getId());
                               consumeDo.setClassName(o.getClassName());
                               consumeDo.setMethodName(o.getMethodName());
                               consumeDo.setStatus(MessageQueueLogsStatusEnum.WAIT.getStatus());
                               if(ObjectUtil.isNotEmpty(batchSize)){
                                   consumeDo.setBatchSize(Integer.valueOf(batchSize));
                               }
                               if(log.isDebugEnabled()){
                                   log.debug("db消息队列执行器-消息消费,请求:{}", JSON.toJSONString(consumeDo));
                               }
                               R r = dbMessageQueueService.consume(consumeDo);
                               if(log.isDebugEnabled()){
                                   log.debug("db消息队列执行器-消息消费,响应:{}", JSON.toJSONString(r));
                               }
                           });
                        });
                    });
                    log.info("db消息队列执行器---线程池执行结束,主题数量:{}", configs.size());
                    //查询需要处理的数据
                    TimeUnit.SECONDS.sleep(Integer.valueOf(executeWaitTime));
                } catch (Exception e) {
                    log.error("db消息队列执行器-周期执行异常", e);
                }
            }
        }).start();
    }

    /**
     * 调用消费者业务处理
     * @param className
     * @param methodName
     * @param supplier
     * @return
     */
    public R invokeConsumer(String className, String methodName, Supplier supplier){
        try {
            Object bean = SpringUtil.getBean(Class.forName(className));
            if(bean == null){
                return R.failed("无效的bean").params("className", className).params("methodName", methodName);
            }
            //反射调用方法，传递logId
            Class<?> aClass = bean.getClass();
            Method method = null;
            for (Method declaredMethod : aClass.getDeclaredMethods()) {
                if(declaredMethod.getName().equals("equals")
                        || declaredMethod.getName().equals("toString") || declaredMethod.getName().equals("hashCode")){
                    continue;
                }
                if(declaredMethod.getName().equals(methodName)){
                    method = declaredMethod;
                    break;
                }
            }
            if(method == null){
                log.error(StrUtil.format("调用消费者业务处理错误,无效的方法{}#{}", className, methodName));
                return R.failed(StrUtil.format("调用消费者业务处理错误,无效的方法{}#{}", className, methodName));
            }
            Object invoke = method.invoke(bean, supplier.get());
            return (R) invoke;
        } catch (Exception e) {
            log.error("调用消费者业务处理异常", e);
        }
        return R.failed("调用消费者业务处理异常").params("className", className).params("methodName", methodName);
    }

}
